/*
 * Created on May 14, 2005
 *
 */
package org.mushroomdb.engine.operation.impl;

import java.util.Iterator;

import org.mushroomdb.catalog.Register;
import org.mushroomdb.engine.operation.AbstractBinaryRelationOperation;
import org.mushroomdb.engine.operation.RelationOperation;
import org.mushroomdb.engine.optimizer.RelationOperationVisitor;
import org.mushroomdb.executionplan.ExecutionPlanElement;
import org.mushroomdb.executionplan.ExecutionPlanElementImpl;


/**
 * @author Matias
 *
 */
public class IntersectionOperation extends AbstractBinaryRelationOperation {

	/**
	 * @param leftRelationOperation
	 * @param rightRelationOperation
	 */
	public IntersectionOperation(RelationOperation leftRelationOperation, RelationOperation rightRelationOperation) {
		super(leftRelationOperation, rightRelationOperation);
	}

	/**
	 * @see org.mushroomdb.engine.operation.RelationOperation#execute()
	 */
	public Iterator execute() {
		final Iterator leftIterator = this.leftRelationOperation.execute();
		
		return new Iterator() {
			private Register foundRegister = null;
			
			public void remove() {
				leftIterator.remove();
			}

			public boolean hasNext() {
				return this.findNext() != null;
			}

			public Object next() {
				Register register = this.findNext();
				this.foundRegister = null;
				return register;
			}
			
			/**
			 * Devuelve el primer registro que coincide.
			 */
			private Register findNext() {
				if (this.foundRegister != null) {
					return this.foundRegister;
				}

				/** Busca un nuevo registro */
				while (leftIterator.hasNext()) {
					Register register = (Register) leftIterator.next();
					Iterator rightIterator = IntersectionOperation.this.rightRelationOperation.execute();
					while (rightIterator.hasNext()) {
						if (IntersectionOperation.this.compares(register,(Register) rightIterator.next())) {
							this.foundRegister = register;
							return this.foundRegister;
						}
					}
				}
				return null;
			}
		};
	}
	
	/**
	 * Compara si dos registros son iguales.
	 * @param leftRegister
	 * @param rightRegister
	 * @return
	 */
	private boolean compares(Register leftRegister, Register rightRegister) {
		if (leftRegister.getColumnCount() != rightRegister.getColumnCount()) {
			return false;
		} else {
			int columnCount = leftRegister.getColumnCount();
			for (int i = 1; i < columnCount; i++) {
				if (!leftRegister.getValue(i).equals(rightRegister.getValue(i))) {
					return false;
				}
			}
			return true;
		}
	}

	/**
	 * toString
	 */
	public String toString() {
		return "(" + this.leftRelationOperation.toString() + //$NON-NLS-1$
			" INTERSECTION " + //$NON-NLS-1$
			this.rightRelationOperation + ")"; //$NON-NLS-1$
	}

	/**
	 * @see org.mushroomdb.engine.operation.RelationOperation#buildExecutionPlan()
	 */
	public ExecutionPlanElement buildExecutionPlan() {
		ExecutionPlanElementImpl executionPlan = new ExecutionPlanElementImpl("INTERSECTION"); //$NON-NLS-1$
		executionPlan.addChild(this.leftRelationOperation.buildExecutionPlan());
		executionPlan.addChild(this.rightRelationOperation.buildExecutionPlan());
		return executionPlan;
	}

	/**
	 * Agregado por Hache para optimizar
	 * @return
	 */
	public void accept(RelationOperationVisitor v) {
		v.visitIntersection(this);
		this.getLeftRelationOperation().accept(v);
		this.getRightRelationOperation().accept(v);
	}
}
